package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
 * 用于统一管理内存中的数据（可能在多线程环境下使用，注意线程安全问题）
 * */
public class MemoryDataCenter {

    //表示交换机 <交换机名字，交换机对象>(保证线程安全问题)
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //表示队列 <队列名字，队列实际对象>
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //<exchangeName,<queueName,binding>> 存储所有的绑定信息
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //<messageID,message对象>
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //<queueName,message对象链表>
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //表示未确认的消息<queueName,<messageId,message>>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //交换机curd
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());

    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }

    public Exchange getExchange(String exchangeName) {
        if(!exchangeMap.containsKey(exchangeName)) {
            System.out.println("exchangeMap中没有:"+exchangeName+"这个交换机！！！查看创建队列的时候是否插入");
            return null;
        }
        return exchangeMap.get(exchangeName);
    }

    //队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName=" + queue.getName());
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName=" + queueName);
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    //绑定
    public void insertBinding(Binding binding) throws MqException {
        //查看exchange是否存在
//         ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.get(binding.getExchangeName());
//         if(bindingMap==null){
//             bindingMap=new ConcurrentHashMap<>();
//         }


        //computeIfAbsent不存在，则创建key
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName()
                , (key) -> new ConcurrentHashMap<>());

        synchronized (bindingMap) {//查到这个队列和交换机是否绑定，并且插入，需要线程安全！
            //如果交换机和队列绑定过，抛出异常
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("【MemoryDataCenter】绑定已经存在 exchangeName=" + binding.getExchangeName() +
                        " queue=" + binding.getQueueName() + " bingKey=" + binding.getBindingKey());
            }
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
    }

    //1、根据exchange name确定唯一binding
    public Binding getBinding(String queueName, String exchangeName) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        Binding binding = bindingMap.get(queueName);
//        if(binding==null){
//            return null;
//        }
        return binding;
    }

    //2、根据exchangeName 获取所有Binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) throws MqException {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        //binding中当前交换机绑定了那些队列？
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        //这些队列里面又binding的queue吗？
        synchronized (bindingMap) {
            if (bindingMap == null) {
                //没有这个binding，直接报错
                throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingMap.remove(binding.getQueueName());
            System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
    }

    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter]新消息添加成功 messageId=" + message.getMessageId());
    }


    //根据ID查询消息
    public Message getMessage(String messageId) throws MqException {
        return messageMap.get(messageId);
    }

    //根据id删除消息
    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) throws MqException {
        //把消息放到对应的队列数据结构中

        //1、根据队列名字，找到消息链表(没有这个链表则创建,computeIfAbsent是线程安全的)（k是传入的key 智力指的是messageId）
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), (k) -> new LinkedList<>());

        //把数据加到message里面
        synchronized (messages) {
            messages.add(message);

            //把消息也插入到总的消息表中(message已经在总消息表中存在也没关系)
            addMessage(message);
            System.out.println("[MemoryDataCenter] 消息被投递到队列中 messageId=" + message.getMessageId());
        }
    }

    //从队列中取消息
    public Message pollMessage(String queueName) throws MqException {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.isEmpty()) {
                return null;
            }

            Message message = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + message.getMessageId());
            return message;
        }
    }

    //获取指定队列中的消息个数
    public int getMessageCount(String queueName) throws MqException {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    //添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) throws MqException {
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAckMap
                .computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId=" + message.getMessageId());

    }

    //删除未确认的消息（消息已经确认）
    public void removeMessageWaitAck(String queueName, Message message) throws MqException {
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if (messages == null) {
            return;
        }
        messages.remove(message.getMessageId());
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId=" + message.getMessageId());

    }

    //获取指定为确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) throws MqException {
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if (messages == null) {
            return null;
        }
        return messages.get(messageId);
    }

    //从硬盘中恢复数据到内存
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //1、清理内存中的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWaitAckMap.clear();

        //2、读取交换机数据
        List<Exchange> exchanges = diskDataCenter.getAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }

        //3、读取队列
        List<MSGQueue> queues = diskDataCenter.getAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }

        //4、读取绑定
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> exchangeAndBindings = bindingsMap.computeIfAbsent(binding.getExchangeName(), (k) -> new ConcurrentHashMap<>());
            exchangeAndBindings.put(binding.getQueueName(), binding);
        }

        //5、读取消息(根据队列，进行枚举获取，加到queueMessage 和messagesMap)
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFormQueue(queue);
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }

        }


        //queueMessageWaitAckMap消息不需要存放在硬盘也不用恢复，因为一旦断开连接，
        // 进行重连，消费者需要消息，就重新从MQ服务器中获取消息即可
    }

}
